package com.css.apps.task.job;

import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.css.apps.task.model.TTask;
import com.css.core.configuration.Environment;
import com.css.db.memcached.MemCachedFactory;
import com.css.util.Md5Util;
import com.css.util.StringHelper;
import com.css.util.UuidUtil;

import net.sf.json.JSONException;
import net.sf.json.JSONObject;

/**
 * 支持集群任务调度的抽象模板类
 * 
 * @author CSS. WeidongWang
 * @since 2014-12-24
 * @version 0.1
 */
public abstract class JobTemplate {
	
	private static Log log = LogFactory.getLog(JobTemplate.class);
	// 执行方式
	public static Integer EXEC_AUTO = 1;
	public static Integer EXEC_MANUAL = 2;
	// 缓存变量
	public static final String JOB_STATE = "job_state";
	public static final String JOB_LIST = "job_list";
	public static final String JOB_RUNNING = "job_running";
	public static final int casTimes = 100;
	protected TTask task;

	/**
	 * 根据实际需求实现该接口，用于生成调度任务列表
	 * 
	 * @return 调度任务列表
	 */
	protected abstract List<?> beforeJob();

	/**
	 * 根据实际需求实现具体的任务执行方法，自行处理任务列表中对象的状态
	 * 
	 * @param obj
	 *            任务列表中的对象
	 */
	protected abstract void doJob(Object obj);

	public String getMd5Key(String key) {
		return Md5Util.MD5Encode(Environment.SYSTEM_NAME + "TaskId:" + task.getUuid() + "TaskKey:" + key);
	}

	/**
	 * 检查任务在当前的触发周期内初始工作是否已经执行
	 * 
	 * @return boolean <br/>
	 *         true: 已初始化 <br/>
	 *         false: 未初始化<br/>
	 */
	public boolean checkInit() {
		String key = getMd5Key(JOB_STATE);
		/**
		 * lState为任务执行信号量，每个集群节点服务器本机保存一份lState，memcached服务器保存一份mState <br/>
		 * 当定时触发器工作时，判断bRet=lState.equals(mState) <br/>
		 * bRet=false ：意味着任务的初始化工作“没有被”集群中的其他节点执行，本机可执行beforeJob() <br/>
		 * bRet=true：意味着任务的初始化工作“已经被”集群中的其他节点执行，本机不执行beforeJob()，赋值lState=mState<br/>
		 */
		String lState = JobMapUtil.getInstance().get(key);
		if (lState == null)
			lState = "";
		boolean bRet = false;
		boolean flag = false;
		for (int i = 1; i <= casTimes; i++) {
			/*
			 * MemcachedItem mi = MemCachedFactory.buildClient().gets(key); if
			 * (mi != null) { String mState = (String) mi.getValue(); if
			 * (StringHelper.isEmpty(mState) || mState.equals(lState)) { lState
			 * = UuidUtil.getUuid(); JobMapUtil.getInstance().set(key, lState);
			 * mi.value = lState; flag = MemCachedFactory.buildClient().cas(key,
			 * mi.value, mi.getCasUnique()); if (flag) { bRet = false; break; }
			 * } else { lState = mState; JobMapUtil.getInstance().set(key,
			 * lState); bRet = true; break; } } else { lState =
			 * UuidUtil.getUuid(); JobMapUtil.getInstance().set(key, lState);
			 * MemCachedFactory.buildClient().set(key, lState); // bRet = false;
			 * // break; continue; }
			 */
			String mState = (String) MemCachedFactory.buildClient().get(key);
			if (mState != null) {
				if (StringHelper.isEmpty(mState) || mState.equals(lState)) {
					lState = UuidUtil.getUuid();
					JobMapUtil.getInstance().set(key, lState);
					flag = MemCachedFactory.buildClient().set(key, lState);
					if (flag) {
						bRet = false;
						break;
					}
				} else {
					lState = mState;
					JobMapUtil.getInstance().set(key, lState);
					bRet = true;
					break;
				}
			} else {
				lState = UuidUtil.getUuid();
				JobMapUtil.getInstance().set(key, lState);
				MemCachedFactory.buildClient().set(key, lState);
				// bRet = false;
				// break;
				continue;
			}
		}
		return bRet;
	}

	public boolean isRunning() {
		String key = getMd5Key(JOB_RUNNING);
		String running = JobMapUtil.getInstance().get(key);
		if (running == null)
			running = "false";
		return running.equals("true");
	}

	public void running() {
		String key = getMd5Key(JOB_RUNNING);
		JobMapUtil.getInstance().set(key, "true");
	}

	public void finished() {
		String key = getMd5Key(JOB_RUNNING);
		JobMapUtil.getInstance().set(key, "false");
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	private void addJobList(List<?> addList) {
		if (addList == null || addList.size() < 1)
			return;
		String key = getMd5Key(JOB_LIST);
		boolean flag = false;
		for (int i = 1; i <= casTimes; i++) {
			/*
			 * MemcachedItem mi = MemCachedFactory.buildClient().gets(key); if
			 * (mi != null) { List jobList = (List) mi.getValue(); if (jobList
			 * == null) jobList = new ArrayList(); jobList.addAll(addList);
			 * mi.value = jobList; flag =
			 * MemCachedFactory.buildClient().cas(key, mi.value,
			 * mi.getCasUnique()); if (flag) break; } else {
			 * MemCachedFactory.buildClient().set(key, addList); break; }
			 */
			List jobList = (List<?>) MemCachedFactory.buildClient().get(key);
			if (jobList != null) {
				jobList.addAll(addList);
				flag = MemCachedFactory.buildClient().set(key, jobList);
				if (flag)
					break;
			} else {
				MemCachedFactory.buildClient().set(key, addList);
				break;
			}
		}
	}

	@SuppressWarnings("rawtypes")
	private boolean hasJob() {
		try {
			String key = getMd5Key(JOB_LIST);
			boolean flag = false;
			for (int i = 1; i <= casTimes; i++) {
				/*
				 * MemcachedItem mi = MemCachedFactory.buildClient().gets(key);
				 * if (mi == null) return false; else { List jobList = (List)
				 * mi.getValue(); if (jobList == null || jobList.size() < 1)
				 * return false; else { System.out.println(jobList.size());
				 * Object obj = jobList.get(0); jobList.remove(0); mi.value =
				 * jobList; flag = MemCachedFactory.buildClient().cas(key,
				 * mi.value, mi.getCasUnique()); if (flag) { doJob(obj); return
				 * true; } } }
				 */
				List jobList = (List) MemCachedFactory.buildClient().get(key);
				if (jobList == null || jobList.size() < 1)
					return false;
				else {
					Object obj = jobList.get(0);
					jobList.remove(0);
					flag = MemCachedFactory.buildClient().set(key, jobList);
					if (flag) {
						doJob(obj);
						return true;
					}
				}
			}
			return false;
		} catch (Exception ex) {
			ex.printStackTrace();
			return false;
		}
	}

	/*
	 * public Object nextJob(String JOB_LIST) { Object item = null; boolean flag
	 * = false; try { String key = getMd5Key(JOB_LIST); for (int i = 1; i <=
	 * casTimes; i++) { //从分布式队列中取数据，即大家（所有节点）从同一个地方取数据 MemcachedItem mi =
	 * MemCachedFactory.buildClient().gets(key); if (mi != null) {
	 * LinkedHashMap<String, Object> jobList = (LinkedHashMap<String, Object>)
	 * mi.getValue(); if (jobList == null || jobList.size() < 1) break;
	 * System.out.println(jobList.size()); // item = jobList.get(0); //
	 * jobList.remove(0); item = getItem(jobList); mi.value = jobList; flag =
	 * MemCachedFactory.buildClient().cas(key, mi.value, mi.getCasUnique()); }
	 * else { flag = true; } if (flag) { break; } } } catch (Exception ex) {
	 * ex.printStackTrace(); } return flag ? item : null; } private Object
	 * getItem(LinkedHashMap<String, Object> linkedHashMap) { Iterator<String> i
	 * = linkedHashMap.keySet().iterator(); Object item = null; if (i.hasNext())
	 * { String key = i.next(); item = linkedHashMap.get(key);
	 * linkedHashMap.remove(key); } return item; }
	 */
	public String getExePara(String key) {
		try {
			if (StringHelper.isEmpty(task.getExePara()))
				return null;
			JSONObject jsonObject = JSONObject.fromObject(task.getExePara());
			return (String) jsonObject.getString(key);
		} catch (JSONException e) {
			e.printStackTrace();
			return null;
		}
	}

	public void execute(TTask task) {
		try {
			this.task = task;
			log.info("任务调度开始执行：task id:" + task.getUuid() + " name:" + task.getName());
			if (isRunning())
				return;
			running();
			switch (Integer.valueOf(task.getJobType())) {
			case 1:
				// 队列任务
				if (!this.checkInit())
					addJobList(beforeJob());
				while (hasJob()) {
				}
				break;
			case 2:
				// 固定任务
				if (this.checkInit())
					return;
				doJob(null);
				break;
			}
		} catch (Exception ex) {
			log.error(ex.getStackTrace());
		} finally {
			finished();
		}
	}
}